package org.zjt.rpc.util.watcher;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import com.alibaba.fastjson.JSONObject;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.zjt.rpc.domain.RegisteNodeData;
import org.zjt.rpc.service.RPCFacade;
import org.zjt.rpc.service.RpcImpl;

public class ZkRegister {
	
	/** zookeeper地址 */
	static final String CONNECT_ADDR = "localhost:12181";
	/** session超时时间 */
	static final int SESSION_OUTTIME = 5000;//ms

	static final String RPC_PARENT_PATH = "/rpc";

	static final String ZK_SEPARATOR = "/";

	/**	保存interface class ：多个对等服务 */
	static final ConcurrentHashMap<String,Set<RegisteNodeData>> REGISTER_ROUTE_MAP = new ConcurrentHashMap();

	public static void watchNodeData(CuratorFramework cf) throws Exception {

		//4 建立一个PathChildrenCache缓存,第三个参数为是否接受节点数据内容 如果为false则不接受
		PathChildrenCache cache = new PathChildrenCache(cf, RPC_PARENT_PATH, true);
		//5 在初始化的时候就进行缓存监听
		cache.start(StartMode.POST_INITIALIZED_EVENT);
		cache.getListenable().addListener(new PathChildrenCacheListener() {
			/**
			 * <B>方法名称：</B>监听子节点变更<BR>
			 * <B>概要说明：</B>新建、修改、删除<BR>
			 * @see PathChildrenCacheListener#childEvent(CuratorFramework, PathChildrenCacheEvent)
			 */
			@Override
			public void childEvent(CuratorFramework cf, PathChildrenCacheEvent event) throws Exception {
				switch (event.getType()) {
					case CHILD_ADDED:
						System.out.println("CHILD_ADDED :" + event.getData().getPath()+"="+event.getData().getData());

						//CHILD_ADDED :/rpc/org.zjt.rpc.service.RPCFacade


						//REGISTER_ROUTE_MAP.put(getFacadeName(event),)
						break;
					case CHILD_UPDATED:
						System.out.println("CHILD_UPDATED :" + event.getData().getPath());


						break;
					case CHILD_REMOVED:
						System.out.println("CHILD_REMOVED :" + event.getData().getPath());


						break;
					default:
						break;
				}
			}
		});

		Thread.sleep(Integer.MAX_VALUE);

	}

	public static CuratorFramework initConnect(){
		//1 重试策略：初试时间为1s 重试10次
		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
		//2 通过工厂创建连接
		CuratorFramework cf = CuratorFrameworkFactory.builder()
				.connectString(CONNECT_ADDR)
				.sessionTimeoutMs(SESSION_OUTTIME)
				.retryPolicy(retryPolicy)
				.build();

		//3 建立连接
		cf.start();
		return cf;
	}

	//   path: RPC_PARENT_PATH/interface/host/
	public static void registerNodeData(Map<String,RegisteNodeData> data ,String host)throws Exception{
		CuratorFramework cf =initConnect();
		for (Map.Entry entry : data.entrySet()){
			Stat stat = cf.checkExists().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey()+ZK_SEPARATOR+host);
			if (stat == null)
			cf.create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey()+ZK_SEPARATOR+host, JSONObject.toJSONString(entry.getValue()).getBytes());
		}
		loadNodeData(cf);
		watchNodeData(cf);
	}

	private static void loadNodeData(CuratorFramework cf) throws Exception{
		List<String> list = cf.getChildren().forPath(RPC_PARENT_PATH);
		if (list == null) return;
		for (String path : list){
			Set<RegisteNodeData> nodeData;
			if (!REGISTER_ROUTE_MAP.contains(path))
				REGISTER_ROUTE_MAP.put(path,new HashSet<>());
			nodeData = REGISTER_ROUTE_MAP.get(path);
			List<String> hosts =cf.getChildren().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+path);
			if (hosts == null) continue;
			for (String host : hosts){
				String content = new String(cf.getData().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+path+ZK_SEPARATOR+host));
				RegisteNodeData registeNodeData = JSONObject.parseObject(content , RegisteNodeData.class);
				registeNodeData.setHost(host);
				nodeData.add(registeNodeData);
			}
		}
	}


	private static String getFacadeName(PathChildrenCacheEvent event) throws Exception{
		return event.getData().getPath().replace(ZK_SEPARATOR +RPC_PARENT_PATH+ZK_SEPARATOR,"");
	}


	public static void main(String[] args)throws Exception{
		Map<String,RegisteNodeData> data = new HashMap<>();

		data.put(RPCFacade.class.getName(),new RegisteNodeData("localhost",8070, RpcImpl.class.getName()));

		registerNodeData(data,"localhost");
	}
}
